package com.lkh.real;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.RestClientBuilder;

import java.util.*;

public class ComputeWeiBoToES {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("zookeeper.connect", "node2:2181,master:2181,node1:2181");
        kafkaProps.setProperty("bootstrap.servers", "node2:9092");
        kafkaProps.setProperty("group.id", "1");
        //构建kafkaSource
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                "wz1",
                new SimpleStringSchema(),
                kafkaProps);
        kafkaConsumer.setStartFromEarliest();
        SingleOutputStreamOperator<String> filter = env
                .addSource(kafkaConsumer)
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        return value.split("\\|").length == 8;
                    }
                });
        //  es 连接地址
        List<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("node1", 9200, "http"));
        ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<String>() {
                    IndexRequest createIndexRequest(String element) {
                        Map<String, String> json = new HashMap<>();

                        String[] split = element.split("\\|");

                        json.put("id", split[0]);
                        json.put("sent_id", split[1]);
                        json.put("user_id", split[2]);
                        json.put("source", split[3]);
                        json.put("reposts_count", split[4]);
                        json.put("comments_count", split[5]);
                        json.put("attitudes_count", split[6]);
                        json.put("text", split[7]);


                        return Requests.indexRequest()
                                .id(split[0])
                                //上面的topic用的wz1，这是创建的索引是wz
                                .index("wz")
                                .source(json);
                    }

                    @Override
                    public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
                        indexer.add(createIndexRequest(element));
                    }
                }
        );

        esSinkBuilder.setRestClientFactory(new RestClientFactory() {
            @Override
            public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
                Header[] headers = new BasicHeader[]{new BasicHeader("Content-Type", "application/json")};
                restClientBuilder.setDefaultHeaders(headers); //以数组的形式可以添加多个header
            }
        });

        filter.print();
        //数据sink到es
        filter.addSink(esSinkBuilder.build());

        env.execute();

    }
}
